package com.etc

import org.apache.spark.sql.types.{DoubleType, StringType, StructField, StructType}
import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.functions._

/**
  * @Auther: Wangcc
  * @Date: 2018/8/24 15:51
  * @Description: 根据每天的用户购买日志 ，销售额
  */
object DailySale {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName("DailySale")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)

    import sqlContext.implicits._

    // 模拟数据
    val userSaleLog = Array(
      "2015-10-01,55.05,1122",
      "2015-10-01,23.15,1133",
      "2015-10-01,15.20,",
      "2015-10-02,56.05,1144",
      "2015-10-02,78.87,1155",
      "2015-10-02,113.02,1123")

    val userSaleLogRDD = sc.parallelize(userSaleLog)

    // 进行有效销售日志的过滤
    val filteredUserSaleLogRDD = userSaleLogRDD.filter(
      log => if (log.split(",").length == 3) true else false)

    val userSaleLogRowRDD = filteredUserSaleLogRDD
      .map { log => Row(log.split(",")(0), log.split(",")(1).toDouble) }

    val structType = StructType(Array(
      StructField("date", StringType, true),
      StructField("sale_amount", DoubleType, true)))


    val userSaleLogDF = sqlContext.createDataFrame(userSaleLogRowRDD, structType)

    val dataFrame = userSaleLogDF.groupBy("date")
      .agg('date, sum('sale_amount))
    dataFrame.foreach {
      ree => println(ree(0) + "," + ree(2))
    }
  }

}
